package com.test.mq;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class AceMQConsumer {
	public static final Logger LOGGER=LoggerFactory.getLogger(AceMQConsumer.class);
	private String topic="orderTopic";
	private String groupName="coupon_orderTopic_cancel_group";
	private String namesrvAddr="192.168.0.105:9876;192.168.0.106:9876";
    private String tag="*";
    private int consumeThreadMin=20;
    private int consumeThreadMax=64;
    
    private IMessageProcessor processor=new CancelOrderProcessor();
    
    public String getTopic() {
		return topic;
	}

	public void setTopic(String topic) {
		this.topic = topic;
	}

	public String getGroupName() {
		return groupName;
	}

	public void setGroupName(String groupName) {
		this.groupName = groupName;
	}

	public String getNamesrvAddr() {
		return namesrvAddr;
	}

	public void setNamesrvAddr(String namesrvAddr) {
		this.namesrvAddr = namesrvAddr;
	}

	public IMessageProcessor getProcessor() {
		return processor;
	}

	public void setProcessor(IMessageProcessor processor) {
		this.processor = processor;
	}
	
	public AceMQConsumer() throws AceMQException{
		if(StringUtils.isBlank(groupName)){
    		throw new AceMQException("groupName is blank!");
    	}
    	if(StringUtils.isBlank(topic)){
    		throw new AceMQException("topic is blank!");
    	}
    	if(StringUtils.isBlank(namesrvAddr)){
    		throw new AceMQException("namesrvAddr is blank!");
    	}
    	DefaultMQPushConsumer consumer=new DefaultMQPushConsumer(this.groupName);
    	consumer.setNamesrvAddr(this.namesrvAddr);
    	try {
			consumer.subscribe(this.topic, this.tag);
			consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
	    	consumer.setConsumeThreadMin(consumeThreadMin);
	    	consumer.setConsumeThreadMax(consumeThreadMax);
	    	AceMessageListener listener=new AceMessageListener();
	    	listener.setMessageProcessor(processor);
	    	consumer.registerMessageListener(listener);
	    	consumer.start();
	    	LOGGER.info(String.format("consumer is start! groupName:{%s},topic:{%s},namesrvAddr:{%s}", this.groupName,this.topic,this.namesrvAddr));
		} catch (MQClientException e) {
			LOGGER.error(String.format("consumer error! groupName:{%s},topic:{%s},namesrvAddr:{%s}", this.groupName,this.topic,this.namesrvAddr,e));
			throw new AceMQException(e);
		}
    	
    }
    public static void main(String[] args) throws AceMQException {
		new AceMQConsumer();
	}
	

}
